package com.vehicle.mq.event;

import com.lmax.disruptor.EventTranslator;
import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;
import com.vehicle.common.redis.manager.CacheManager;
import com.vehicle.common.util.thread.pool.ThreadPoolManager;
import com.vehicle.mq.message.BaseMessage;
import com.vehicle.mq.message.MessageEvent;
import com.vehicle.mq.queue.EventQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;

/**
 * Created by yantingbin on 2017/12/5.
 */
public class EventPublishThread {
    private static final Logger LOGGER = LoggerFactory.getLogger(EventPublishThread.class);

    private final Executor EXECUTOR = ThreadPoolManager.I.getEventPublishExecutor();

    private final EventQueue eventQueue;

    public EventPublishThread(EventQueue eventQueue) {
        this.eventQueue = eventQueue;
    }

    private final EventTranslatorOneArg<MessageEvent, BaseMessage> EVENTTRANSLATOR = new EventTranslatorOneArg<MessageEvent, BaseMessage>() {
        @Override
        public void translateTo(MessageEvent messageEvent, long l, BaseMessage baseMessage) {
            messageEvent.setBaseMessage(baseMessage);
        }
    };

    public void make(final RingBuffer<MessageEvent> ringBuffer) {
        EXECUTOR.execute(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    BaseMessage baseMessage = eventQueue.next();
                    ringBuffer.publishEvent(EVENTTRANSLATOR, baseMessage);
                }
            }
        });
    }
}
